package flink.examples.sql._03.source_sink.table.redis.v2.sink;

import org.apache.flink.api.common.serialization.SerializationSchema;
import org.apache.flink.streaming.connectors.redis.RedisSink;
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisConfigBase;
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisMapper;
import org.apache.flink.table.connector.ChangelogMode;
import org.apache.flink.table.connector.format.EncodingFormat;
import org.apache.flink.table.connector.sink.DynamicTableSink;
import org.apache.flink.table.connector.sink.SinkFunctionProvider;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.types.DataType;
import org.apache.flink.types.RowKind;
import org.apache.flink.util.Preconditions;

import flink.examples.sql._03.source_sink.table.redis.mapper.SetRedisMapper;
import flink.examples.sql._03.source_sink.table.redis.options.RedisWriteOptions;

public class RedisDynamicTableSink implements DynamicTableSink {

    protected final DataType physicalDataType;
    protected final RedisWriteOptions redisWriteOptions;
    private final EncodingFormat<SerializationSchema<RowData>> encodingFormat;

    public RedisDynamicTableSink(
            DataType physicalDataType,
            RedisWriteOptions redisWriteOptions,
            EncodingFormat<SerializationSchema<RowData>> encodingFormat) {
        this.physicalDataType = Preconditions.checkNotNull(
                physicalDataType, "Physical data type must not be null.");
        this.redisWriteOptions = redisWriteOptions;
        this.encodingFormat = encodingFormat; // Store the encoding format.
    }

    @Override
    public ChangelogMode getChangelogMode(ChangelogMode requestedMode) {
        // UPSERT mode
        ChangelogMode.Builder builder = ChangelogMode.newBuilder();
        for (RowKind kind : requestedMode.getContainedKinds()) {
            if (kind != RowKind.UPDATE_BEFORE) {
                builder.addContainedKind(kind);
            }
        }
        return builder.build();
    }

    @Override
    public SinkRuntimeProvider getSinkRuntimeProvider(Context context) {
        // Create SerializationSchema from the encoding format
        SerializationSchema<RowData> serializationSchema = encodingFormat.createRuntimeEncoder(context, physicalDataType);

        // Configure Redis
        FlinkJedisConfigBase flinkJedisConfigBase = new FlinkJedisPoolConfig.Builder()
                .setHost(this.redisWriteOptions.getHostname())
                .setPort(this.redisWriteOptions.getPort())
                .build();

        // Use the SerializationSchema in the RedisMapper
        RedisMapper<RowData> redisMapper = null;

        switch (this.redisWriteOptions.getWriteMode()) {
            case "string":
                redisMapper = new SetRedisMapper(serializationSchema);
                break;
            default:
                throw new RuntimeException("其他类型 write mode 请自定义实现");
        }

        // Return the sink function provider
        return SinkFunctionProvider.of(new RedisSink<>(
                flinkJedisConfigBase,
                redisMapper
        ));
    }

    @Override
    public DynamicTableSink copy() {
        // Return a copy of the sink with the same properties.
        return new RedisDynamicTableSink(this.physicalDataType, this.redisWriteOptions, this.encodingFormat);
    }

    @Override
    public String asSummaryString() {
        return "Redis Dynamic Table Sink";
    }
}